package com.example.spark.rdd;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.function.Consumer;

public class RDD_FlatMap {

    public static void main(String[] args) {

        //创建配置对象
        SparkConf conf = new SparkConf();
        conf.setMaster("local[*]");
        conf.setAppName("RDDFlatMap");

        //创建SparkContext
        JavaSparkContext jsc = new JavaSparkContext(conf);

        //逻辑代码
        //源数据
        List<List<String>> lists = new ArrayList<>();
        lists.add(Arrays.asList("1","2"));
        lists.add(Arrays.asList("3","4"));

        //List<String> list ={"1","2","3","4"}

        JavaRDD<List<String>> rdd1 = jsc.parallelize(lists);
        JavaRDD<String> rdd2 = rdd1.flatMap(new FlatMapFunction<List<String>, String>() {
            @Override
            public Iterator<String> call(List<String> strings) throws Exception {
                return strings.iterator();
            }
        });

        List<String> collect = rdd2.collect();
        collect.forEach(System.out::println);
//        collect.forEach(new Consumer<String>() {
//            @Override
//            public void accept(String s) {
//                System.out.println(s);
//            }
//        });

  System.out.println("=================================");

        //拆分文本里的单词作为集合的元素
        JavaRDD<String> rdd3 = jsc.textFile("input/2.txt");
        JavaRDD<String> rdd4 = rdd3.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String line) throws Exception {
                String[] words = line.split(" ");
                List<String> wordList = Arrays.asList(words);

                return wordList.iterator();
            }
        });
        //拿到RDD的数据集合并遍历输出
        rdd4.collect().forEach(System.out::println);


        //释放资源
        jsc.stop();


    }

}
